package task

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"sieve_engine/execute/whatsapp/log"

	"github.com/gogf/gf/container/gqueue"
	"go.uber.org/zap"

	// "sieve_engine/execute/whatsapp/log"
	"sieve_engine/lib/config"
)

var (
	whatsAppTokenConfig = WhatsAppTokenConfig{
		ServerIp:       *config.String("whatsapp.token.server_ip", "", "whatsapp token request server_ip"),
		RequestCount:   int32(*config.Int("whatsapp.token.request_count", 1, "whatsapp token request count")),
		SyncInterval:   int32(*config.Int("whatsapp.token.sync_interval", 60000, "whatsapp token sync interval")),
		HttpTimeout:    int32(*config.Int("whatsapp.token.http_timeout", 5, "whatsapp token http timeout")),
		LocalCacheFile: *config.String("whatsapp.token.local_cache_file", "/data/sieve_engine/task/tokens.json", "whatsapp token local cache file"),
	}
)

type WhatsAppTokenConfig struct {
	ServerIp       string `json:"server_ip"  mapstructure:"server_ip"`
	RequestCount   int32  `json:"request_count"  mapstructure:"request_count"`
	SyncInterval   int32  `json:"sync_interval"  mapstructure:"sync_interval"`
	HttpTimeout    int32  `json:"http_timeout"  mapstructure:"http_timeout"`
	LocalCacheFile string `json:"local_cache_file"   mapstructure:"local_cache_file"`
}

type whatsAppTokenServiceProvider struct {
	closeChan          chan int
	lockMtx            sync.Mutex
	subscriber         whatsAppTokenServiceProviderInterface
	httpClient         *http.Client
	syncer             *sync.WaitGroup
	logger             *zap.Logger
	updateTokenCh      chan int
	tokens             map[string]*WasTokenBean
	initialized        atomic.Bool
	cacheTokenQueue    *gqueue.Queue
	activeTokenCount   atomic.Int32
	token_trace_logger *zap.Logger
}

type whatsAppTokenServiceProviderInterface interface {
	NotifyPulled(token *WasTokenBean)
}

var WhatsAppTokenProvider *whatsAppTokenServiceProvider
var whatsappTopkenServiceTokenOnce sync.Once

func NewWhatsAppTokenServiceProvider(close chan int, consumer whatsAppTokenServiceProviderInterface) *whatsAppTokenServiceProvider {
	whatsappTopkenServiceTokenOnce.Do(func() {
		WhatsAppTokenProvider = &whatsAppTokenServiceProvider{
			closeChan:          close,
			subscriber:         consumer,
			httpClient:         &http.Client{Timeout: time.Second * time.Duration(whatsAppTokenConfig.HttpTimeout)},
			logger:             log.CreateLogger(close, "whatsapp_token"),
			updateTokenCh:      make(chan int, whatsAppTokenConfig.RequestCount),
			tokens:             make(map[string]*WasTokenBean),
			cacheTokenQueue:    gqueue.New(),
			token_trace_logger: log.CreateLogger(close, "whatsapp_token_tracer"),
		}
		WhatsAppTokenProvider.initialized.Store(false)
		WhatsAppTokenProvider.token_trace_logger.Info("ConfigTokenCount", zap.Int32("count", whatsAppTokenConfig.RequestCount))

		WhatsAppTokenProvider.ScheduleBackground()
	})

	return WhatsAppTokenProvider
}

func (provider *whatsAppTokenServiceProvider) ScheduleBackground() {

	go func() {
		for {
			select {
			case <-provider.closeChan:
				return
			case <-provider.updateTokenCh:
				provider.updateServerToken()
			}
		}
	}()
	for i := 0; i < int(whatsAppTokenConfig.RequestCount); i++ {
		provider.updateTokenCh <- 1
	}
}

func (provider *whatsAppTokenServiceProvider) syncToken() {
	provider.updateTokenCh <- 1
}

func (provider *whatsAppTokenServiceProvider) updateServerToken() {
	provider.logger.Info("updateServerToken", zap.Time("update_time", time.Now()))
	tokenBeans, err := provider.requestRemoteHttpServerApi()
	if err != nil {
		provider.logger.Error("updateServerToken failed", zap.Error(err))
		provider.syncToken()
		return
	}
	for _, tbean := range tokenBeans {
		provider.logger.Info("tokenbean", zap.Reflect("bean", tbean))
		// necessary
		tbean.FixData()

		tbean.Logger = provider.token_trace_logger

		provider.lockMtx.Lock()
		if _, ok := provider.tokens[tbean.Phone]; ok {
			tbean.Logger.Info("RemoteTokenRepeated", zap.String("phone", tbean.Phone))
			provider.lockMtx.Unlock()
			provider.syncToken()
			continue
		}
		provider.lockMtx.Unlock()

		if tbean.ClientStaticKeypair == "" || tbean.ClientStaticKeypair == "==" {
			tbean.Logger.Info("TokenInvalid", zap.String("phone", tbean.Phone))
			provider.syncToken()
			continue
		}

		if provider.subscriber != nil {
			provider.subscriber.NotifyPulled(tbean)
		}
	}
}

func (provider *whatsAppTokenServiceProvider) readLocalCache() (error, []*WasTokenBean) {
	provider.logger.Info("readLocalCache")
	bytes, err := os.ReadFile(whatsAppTokenConfig.LocalCacheFile)
	if err != nil {
		provider.logger.Info("cache file empty or not exist")
		return nil, nil
	}
	var tokens []*WasTokenBean
	err = json.Unmarshal(bytes, &tokens)
	if err != nil {
		return errors.New(fmt.Sprintf("unmarshal token cache file %s failed", whatsAppTokenConfig.LocalCacheFile)), nil
	}
	// debug
	provider.token_trace_logger.Info("readLocalCache", zap.Int("count", len(tokens)))
	return nil, tokens
}

func (provider *whatsAppTokenServiceProvider) tellToken(token *WasTokenBean, valid bool) {
	if valid {
		token.Logger.Info("tellTokenValid", zap.String("phone", token.Phone))
	} else {
		token.Logger.Error("tellTokenInValid", zap.String("phone", token.Phone))
	}
	provider.lockMtx.Lock()
	defer provider.lockMtx.Unlock()
	if valid {
		provider.tokens[token.Phone] = token
	} else {
		delete(provider.tokens, token.Phone)
		defer func() {
			if !valid {
				// request remote
				provider.syncToken()
			}
		}()
	}

	var tokendata []*WasTokenBean

	var tokens []string

	for _, v := range provider.tokens {
		tokendata = append(tokendata, v)
		tokens = append(tokens, v.Phone)
	}

	err := os.WriteFile(whatsAppTokenConfig.LocalCacheFile, []byte(marshal(tokendata)), 0777)
	if err != nil {
		provider.logger.Error("write token cache file failed", zap.Error(err))
		return
	} else {
		provider.logger.Info("write token cache file success", zap.String("time", time.Now().Format("2006-01-02 15:04:05")))
	}

	provider.token_trace_logger.Info("SyncValidTokenParts", zap.String("tokens", marshal(tokens)))

	provider.logger.Info("SyncValidTokenCount", zap.Int("token", len(provider.tokens)))
	// debug
	provider.token_trace_logger.Info("SyncValidTokenCount", zap.Int("token", len(provider.tokens)))
}

// requestRemoteHttpServerApi
func (provider *whatsAppTokenServiceProvider) requestRemoteHttpServerApi() ([]*WasTokenBean, error) {
	if !provider.initialized.Load() {
		provider.initialized.Store(true)
		err, tokenBeans := provider.readLocalCache()
		if err != nil {
			provider.logger.Error("read token cache failed", zap.Error(err))
		} else {
			for _, t := range tokenBeans {
				provider.cacheTokenQueue.Push(t)
			}
		}
	}

	if provider.cacheTokenQueue.Size() > 0 {
		tb := provider.cacheTokenQueue.Pop().(*WasTokenBean)
		tb.Logger = provider.token_trace_logger
		tb.FixData()
		tb.Logger.Info("FromCache", zap.String("phone", tb.Phone))
		return []*WasTokenBean{tb}, nil
	}

	provider.logger.Info("getRemoteTokens")
	// debug
	provider.token_trace_logger.Info("getRemoteTokens")
	resp, err := provider.httpClient.Get(provider.getRequestUrl())
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.New(fmt.Sprintf("get failed,%s", resp.Status))
	}
	tokenResp := new(ServerWasTokenResponse)
	bytes, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(bytes, tokenResp)
	if err != nil {
		return nil, err
	}

	if tokenResp.Code != 200 {
		return nil, errors.New(fmt.Sprintf("token server return code=%d,message:%s", tokenResp.Code, tokenResp.Message))
	}

	for _, tb := range tokenResp.Data {
		tb.Logger = provider.token_trace_logger
		tb.FixData()
		tb.Logger.Info("FromRemote", zap.String("phone", tb.Phone))
	}

	return tokenResp.Data, nil
}

// getRequestUrl
func (provider *whatsAppTokenServiceProvider) getRequestUrl() string {
	// @todo 替换config.SieveConfig.Token.RequestCount 为 1
	// 每次请求一个
	return fmt.Sprintf(fmt.Sprintf("%s?num=%d", whatsAppTokenConfig.ServerIp, 1))
}

func marshal(v interface{}) string {
	ret, _ := json.Marshal(v)
	return string(ret)
}
